package com.binarycat.util.arithmetic;

import java.util.List;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.math3.stat.StatUtils;
import org.springframework.util.CollectionUtils;

import rx.Observable;
import rx.subjects.BehaviorSubject;

/**
 * 滑动窗口
 * 
 * @author jjb
 * @date 2018/07/22
 */

public class SlidingWindow {
    private BehaviorSubject<Double> behaviorSubject;
    private Observable<Observable<Double>> observable;
    /**
     * 当前时间窗内的数组数据
     */
    private List<Double> dataList;

    /**
     * 
     * @param timespan 时间值
     * @param unit 时间单位
     */
    public SlidingWindow(long timespan, TimeUnit unit) {
        behaviorSubject = BehaviorSubject.create();
        observable = behaviorSubject.window(timespan, unit);
        startSubscribe();
    }
    public SlidingWindow() {
        behaviorSubject = BehaviorSubject.create();
        observable = behaviorSubject.window(1, TimeUnit.SECONDS);
        startSubscribe();
    }

    /**
     * 一秒内的滑动窗口类
     * 
     * @return
     */
    public static SlidingWindow oneSecondSlidingWindow() {
        return new SlidingWindow(1, TimeUnit.SECONDS);
    }

    /**
     * 一分钟内的滑动窗口类
     * 
     * @return
     */
    public static SlidingWindow oneMinuteSlidingWindow() {
        return new SlidingWindow(1, TimeUnit.MINUTES);
    }

    /**
     * 五分钟内的滑动窗口类
     * 
     * @return
     */
    public static SlidingWindow fiveMinuteSlidingWindow() {
        return new SlidingWindow(5, TimeUnit.MINUTES);
    }

    /**
     * 十五分钟内的滑动窗口类
     * 
     * @return
     */
    public static SlidingWindow fifteenMinuteSlidingWindow() {
        return new SlidingWindow(15, TimeUnit.MINUTES);
    }

    /**
     * 指定时间内的滑动窗口类
     * 
     * @param timespan 时间值
     * @param unit 时间单位
     * @return
     */
    public static SlidingWindow specifiedSlidingWindow(long timespan, TimeUnit unit) {
        return new SlidingWindow(timespan, unit);
    }

    /**
     * 向滑动窗口中添加数据
     * 
     * @param value
     */
    public void addData(double value) {
        behaviorSubject.onNext(value);
    }

    /**
     * 向滑动窗口中添加数据
     * 
     * @param value
     */
    public void addData(int value) {
        behaviorSubject.onNext(Double.valueOf(value));
    }

    private void startSubscribe() {
        observable.subscribe(v -> {
            v.asObservable().toList().subscribe(dataArray -> {
                dataList = dataArray;
            });
        });
    }

    /**
     * 时间窗口中的数据平均数
     * 
     * @return
     */
    public double getDataListMean() {
        if (CollectionUtils.isEmpty(dataList)) {
            return 0;
        }
        return StatUtils.mean(coverToDoubleArray());

    }

    /**
     * 时间窗口中的数据最大值
     * 
     * @return
     */
    public double getDataListMax() {
        if (CollectionUtils.isEmpty(dataList)) {
            return 0;
        }
        return StatUtils.max(coverToDoubleArray());

    }

    /**
     * 时间窗口中的数据最小值
     * 
     * @return
     */
    public double getDataListMin() {
        if (CollectionUtils.isEmpty(dataList)) {
            return 0;
        }
        return StatUtils.min(coverToDoubleArray());

    }

    /**
     * 时间窗口中的百分位数据
     * 
     * @return
     */
    public double getDataListPercentile(double percentile) {
        if (CollectionUtils.isEmpty(dataList)) {
            return 0;
        }
        return StatUtils.percentile(coverToDoubleArray(), percentile);

    }

    /**
     * @return
     */
    private double[] coverToDoubleArray() {
        Double[] array1 = dataList.stream().toArray(Double[]::new);
        double[] array2 = ArrayUtils.toPrimitive(array1);
        return array2;
    }

    public List<Double> getDataList() {
        return dataList;
    }

    public void setDataList(List<Double> dataList) {
        this.dataList = dataList;
    }
}
